package job.gfp.ranks.count_seeds;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;

import mapred.SimpleJob;
import mapred.config.Compress;
import mapred.config.JobConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.RunningJob;

@SuppressWarnings("unchecked")
public class CountSeeds extends SimpleJob {

    public CountSeeds(String... settings) throws IOException {
        super("Count Seeds", settings);

        this.forceUserSetting("-reduce", 0);
    }

    @Override
    protected Class<? extends Reducer> getCombinerClass() {
        return null;
    }

    @Override
    protected Compress getCompress() {
        return Compress.GZIP;
    }

    @Override
    protected String[] getInputPaths(FileSystem fs) throws IOException {
        return new String[] { this.getUserSetting("-linkgraph") };
    }

    @Override
    protected Class<? extends Mapper> getMapperClass() {
        return CountSeeds_Mapper.class;
    }

    @Override
    protected String getOutputPath(FileSystem fs) throws IOException {
        return this.getUserSetting("-output") + "/CountSeeds";
    }

    @Override
    protected Class<? extends Reducer> getReducerClass() {
        return null;
    }

    @Override
    protected int getReducerNumber() {
        return 0;
    }

    @Override
    protected void setUp(JobConfig conf, FileSystem fs) throws IOException {
        conf.set("seeds", this.getUserSetting("-seeds"));
    }

    @Override
    protected void tearDown(JobConfig conf, FileSystem fs, RunningJob job)
            throws IOException {
        long websize = this.getCounter(job, "WebSize", "Number");
        long seeds = this.getCounter(job, "Seeds", "Number");

        double times = 1.0 * websize / seeds;

        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fs
                .create(new Path(this.getUserSetting("-seeds") + ".times"),
                        true)));
        bw.write(String.valueOf(times));
        bw.newLine();
        bw.flush();
        bw.close();

        System.out.println("Times = " + times);
    }
}
